package com.hzgj.bcl.soa.merge;

import com.hzgj.bcl.redis.RedisClient;
import com.hzgj.bcl.stats.Metric;
import com.hzgj.bcl.stats.StatsHandler;
import com.hzgj.bcl.stats.TimeFlowCounter;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Tuple;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by Administrator on 2016/8/18.
 */

public class Merge {
    private Logger logger = LoggerFactory.getLogger(getClass());
    private RedisClient redisClient;
    @Setter
    private long intervalSpan = 1;//默认5分钟
    private String key;
    private AtomicBoolean run = new AtomicBoolean(true);
    Metric metric;
//    public Cache<Long, Metric> bucket =
//            CacheBuilder.newBuilder()
//                    .expireAfterWrite(intervalSpan, TimeUnit.MINUTES)
//                    .maximumSize(1)
//                    .removalListener((RemovalListener<Long, Metric>) notification -> {
//                        logger.info("remove item time=" + new Date(notification.getValue().getTarget()));
//                        StatsHandler.handleDataWithRedis(redisClient, key, notification.getValue());
//                    }).build();

    SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyy-MM-dd HH:mm");

    public Merge(String key, RedisClient redisClient) {
        this.key = key;
        this.redisClient = redisClient;
        metric = new Metric(0, new TimeFlowCounter(intervalSpan,
                (innerMetric, time) -> {
                    StatsHandler.handleDataWithRedis(redisClient, key, innerMetric, time);
                    logger.info("export data key {} in {}", key, simpleDateFormat.format(new Date(time)));
                    innerMetric.clean();
                }));
    }

    public boolean doIt(long start, long end) {

        Set<Tuple> vt = redisClient.zrangeByScoreWithScores(key, start, end);
        if (vt.size() == 0) {
            return true;
        }

        for (Tuple tuple : vt) {
            metric.update(tuple.getElement(), (long) tuple.getScore());
        }
        redisClient.zremrangeByScore(key, start, end);
        return true;
    }

    public void stop() {
        try {
            StatsHandler.handleDataWithRedis(redisClient, key, metric);
            run.compareAndSet(true, false);
        } catch (Exception e) {

        }
    }
}
